package com.disruptor.sample._06multi.run;

import com.disruptor.sample._01base.consumer.LongEventHandler;
import com.disruptor.sample._01base.factory.LongEventFactory;
import com.disruptor.sample._06multi.producer.LongEventProducerByMulti;
import com.disruptor.sample.common.dto.LongEvent;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class LongEventMainByMulti {

  private static ExecutorService PRODUCER_POOL = Executors.newCachedThreadPool();

  private static ThreadFactory CONSUMER_THREAD_FACTORY;

  private static int BUFFER_SIZE = 1024;

  static {
    CONSUMER_THREAD_FACTORY = new ThreadFactory() {
      int i = 0;
      @Override
      public Thread newThread(Runnable r) {
        return new Thread(r, "processorThread" + (i++));
      }
    };
  }

  public static void main(String[] args) throws Exception {
    // 模拟生产者线程数量
    int producerThreadNum = 5;
    // 模拟每个生产者生产消息条数
    int producerDataNum = 2;

    Disruptor<LongEvent> disruptor = new Disruptor<>(new LongEventFactory(), BUFFER_SIZE, CONSUMER_THREAD_FACTORY);

    // 2 个消费者
    disruptor.handleEventsWith(new LongEventHandler(), new LongEventHandler());
    disruptor.start();

    RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
    List<LongEventProducerByMulti> longEventProducers = new ArrayList<>();
    for (int i = 0; i < producerThreadNum; i++) {
      longEventProducers.add(new LongEventProducerByMulti(ringBuffer));
    }

    long start = System.currentTimeMillis();

    CountDownLatch latch = new CountDownLatch(producerThreadNum);

    for (int i = 0; i < producerThreadNum; i++) {
      int finalI = i;
      PRODUCER_POOL.execute(() -> {
        for (long l = 0; l < producerDataNum; l++) {
          longEventProducers.get(finalI).onData(l);
        }
        log.info("latch.count = {}, count --", latch.getCount());
        latch.countDown();
      });
    }

    latch.await();

    long end = System.currentTimeMillis();
    log.info("Disruptor 花费时间: {} ms", (end - start));
  }

}
